package com.example.flink;

import com.example.base.exception.CustomException;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;

import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;

import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.util.Collector;


/**
 * 批处理单词计数器
 * @author wanghongyu02
 * @date 2024-08-13 17:02
 */
public class BatchWordCount {

    public static void main(String[] args) throws Exception {

        try {
            // 1. 创建执行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

            // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
            // 这里不是从resource开始找，是从项目目录往下找，项目目录是otaru不是otaru/back
            // 直接调用执行环境的 readTextFile 方法，可以从文件中读取数据。
            DataSource<String> lineDataSource = env.readTextFile("flink\\words.txt");

            // 3. 转换数据格式
            FlatMapOperator<String,Tuple2<String, Long>> wordAndOne = lineDataSource.flatMap( //调用flatmap方法可以对一行文字进行分词转换
                    (String line, Collector<Tuple2<String, Long>> out) -> { //Tuple2是Flink自带的二元组数据类型
                        String[] words = line.split(" "); //将每行文本进行分词
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1L)); // 将分词结果转换成二元组类型
                        }
                    })
                    .returns(Types.TUPLE(Types.STRING, Types.LONG)); //当 Lambda 表达式使用 Java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息

            // 4. 按照 word 进行分组
            UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);

            // 5. 分组内聚合统计
            AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);

            // 6. 打印结果
            sum.print();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}